package com.kettle.service.kettle.impl;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import com.kettle.utils.CollectionUtil;
import com.kettle.utils.KettleLogUtil;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobEntryResult;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.filerep.KettleFileRepository;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaDataCombi;
import org.pentaho.di.trans.step.StepStatus;
import org.springframework.stereotype.Service;

import com.kettle.model.kettle.Kettlemap;
import com.kettle.service.kettle.KettlesendService;

@Service("kettlesendService")
public class KettlesendServiceImpl implements KettlesendService {
	
	/**
	 * 执行 JOB
	 * @param repository kettle数据库资源库
	 * @param job_name JOB 名称
	 * @param filepath JOB 工作文件夹
	 * @param variableMap 变量Map
	 * @param parameterValueMap 参数Map
	 * @throws KettleException
	 */
	@Override
	public Kettlemap execJob(KettleFileRepository repository,String job_name,String filepath,
			Map<String, Object> variableMap, 
			Map<String, Object> parameterValueMap
			) throws KettleException {
		Kettlemap kmap=new Kettlemap();
		//System.setProperty("KETTLE_DEFAULT_SERVLET_ENCODING", StandardCharsets.UTF_8.name());
		//调整 king
		//RepositoryDirectoryInterface directory = repository.loadRepositoryDirectoryTree();
		RepositoryDirectoryInterface directory = repository.findDirectory(filepath==null?"/":filepath);
		JobMeta jobMeta = repository.loadJob(repository.getJobId(job_name, directory), null);

		//JobMeta jobMeta02 = repository.loadJob(job_name, directory, null, null ) ;

		//JobMeta jobMeta = repository.loadJob(job_name,dir,new ProgressNullMonitorListener(),null);
		Job job = new Job(repository, jobMeta);
		// 开启进程守护
		job.setDaemon(true);
		job.setLogLevel(LogLevel.BASIC);//MINIMAL 有时候获取不到日志信息
		// 设置参数
		if (CollectionUtil.isNotEmpty(parameterValueMap)) {
			Set<Entry<String, Object>> entrySet = parameterValueMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				job.setParameterValue(entry.getKey(), entry.getValue().toString());
			}
		}
		// 设置变量 java代码中变量会覆盖kettle里面设置的变量
		if (CollectionUtil.isNotEmpty(variableMap)) {
			Set<Entry<String, Object>> entrySet = variableMap.entrySet();
			for (Entry<String, Object> entry : entrySet) {
				job.setVariable(entry.getKey(), entry.getValue().toString());
			}
		}
		// 启动执行指定的job
		job.start();
		job.waitUntilFinished();// 等待job执行完
		// 执行完成后获取日志
		String logText = KettleLogUtil.getLogText(job.getLogChannelId(), true, job.getLogDate().getTime());
		// 判断执行过程中是否有错误
		if (job.getErrors() > 0) {

		}
		job.setFinished(true);
		String kstlog="";
		if (job.isFinished()) {
			List<JobEntryResult> jobEntryResults = job.getJobEntryResults();
			for (JobEntryResult jobEntryResult : jobEntryResults) {
				kstlog+="执行步骤=" + jobEntryResult.getJobEntryName() + ",执行状态=" + jobEntryResult.getResult().getResult()+"\n";
				//int kcount=jobEntryResult.getResult().getRows().size();//获取记录数据
			}
		}
		kmap.setSuccess(job.getResult().getResult());//总结果
		kmap.setZxbzlog(kstlog);//步骤日志
		kmap.setLogs(job.getResult().getLogText());//总日志
		// 执行 打印的日志 等全在这个xml中
		System.out.println("------------------------------------------------------------");
		//String kxml=job.getResult().getBasicXml();
		job.stopAll();
		return kmap;
	}
	
	/**
	 * 执行 转换
	 * 
	 * @param repository 文件资源库
	 * @param tranName 转换名称
	 * @param variableMap 参数Map
	 * @param parameterValueMap 参数Map
	 * @throws KettleException
	 */
	@Override
	public Kettlemap execTran(KettleFileRepository repository, String tranName, 
			Map<String, Object> variableMap, 
			Map<String, Object> parameterValueMap){
		Kettlemap kmap=new Kettlemap();
		System.setProperty("KETTLE_DEFAULT_SERVLET_ENCODING", StandardCharsets.UTF_8.name());
		Trans trans=null;
		// 根据指定的字符串路径 找到目录
		try {
			RepositoryDirectoryInterface dir= repository.findDirectory("/");
			// 在 指定目录下找转换
			//TransMeta transMeta = repository.loadTransformation(tranName, dir, new ProgressNullMonitorListener(), true, "1.0");
			//start 获取id king
			ObjectId id = repository.getTransformationID(tranName, dir);
			TransMeta transMeta = repository.loadTransformation(id, null);
	        //创建ktr元对象
	        //TransMeta transMeta = repository.loadTransformation(tranName, dir, null, true, null ) ;
			
			//end 获取id king
			// 设置参数
			if (parameterValueMap != null) {
				Set<Entry<String, Object>> entrySet = parameterValueMap.entrySet();
				for (Entry<String, Object> entry : entrySet) {
					transMeta.setParameterValue(entry.getKey(), entry.getValue().toString());
				}
			}
			// 设置变量 java代码中变量会覆盖kettle里面设置的变量
			if (variableMap != null) {
				Set<Entry<String, Object>> entrySet = variableMap.entrySet();
				for (Entry<String, Object> entry : entrySet) {
					transMeta.setVariable(entry.getKey(), entry.getValue().toString());
				}
			}
			transMeta.setCapturingStepPerformanceSnapShots(true);
			trans = new Trans(transMeta); 
			trans.setLogLevel(LogLevel.BASIC);
	        trans.setMonitored(true);
			trans.setInitializing(true);
			trans.setPreparing(true);
			trans.setRunning(true);
			trans.setSafeModeEnabled(true);
			trans.execute(null);// 执行trans
			trans.waitUntilFinished();// 等待直到数据结束
			//LogChannelInterface log = trans.getLogChannel();
			//log.logBasic(BaseMessages.getString(PKG, "Trans.Log.DispacthingStartedForTransformation", trans.getFilename()));
			// 几个步骤
			List<StepMeta> hopsteps = transMeta.getTransHopSteps(false);
			for (int i = 0; i < hopsteps.size(); i++) {
				StepMeta thisStep = hopsteps.get(i);
				thisStep.getName();// 步骤名称
			}
			// 每个步骤名称
			// StepMetaDataCombi
			List<StepMetaDataCombi> steps = trans.getSteps();
			for (StepMetaDataCombi stepMetaDataCombi : steps) {
				System.out.println(ToStringBuilder.reflectionToString(stepMetaDataCombi));
			}
			// kettle 界面执行面板的 步骤度量
			String kstlog="";
			for (int i = 0; i < trans.nrSteps(); i++) {
				StepInterface baseStep = trans.getRunThread(i);
				StepStatus stepStatus = new StepStatus(baseStep);
				kstlog+="步骤名称=" + stepStatus.getStepname() + ",错误信息=" + stepStatus.getErrors() + ",执行状态=" + stepStatus.getStatusDescription()+"\n";
			}
			kmap.setZxbzlog(kstlog);//步骤日志
			kmap.setLogs(trans.getResult().getLogText());//总日志
			// 转换里面箭头指向
			//List<RowSet> rowsets = trans.getRowsets();
			// 转换里面箭头指向 end
			if (trans.getErrors() > 0) {
				kmap.setSuccess(false);
			}else{
				kmap.setSuccess(true);
			}
		} catch (KettleException e) {
			kmap.setSuccess(false);
			e.printStackTrace();
		}finally{
			if(trans!=null){
			     trans.stopAll();
			}
		}
		return kmap;
	}

}
